#include "raft.h"
#include <boost/archive/text_iarchive.hpp>
#include <boost/archive/text_oarchive.hpp>
#include <memory>
#include "config.h"
#include "util.h"
#include <Logger_v2.h>

// 该函数是节点作为Flower时，为了响应Leader的AppendEntries请求而调用的
void Raft::AppendEntries1(const raftRpcProctoc::AppendEntriesArgs* args, raftRpcProctoc::AppendEntriesReply* reply) {
    // std::lock_guard<std::mutex> locker(m_mtx);
    reply->set_appstate(AppNormal);  // 能接收到代表网络是正常的

    // 不同的人收到AppendEntries的反应是不同的，要注意无论什么时候收到rpc请求和响应都要检查term

    if (args->term() < m_currentTerm) {
        reply->set_success(false);
        reply->set_term(m_currentTerm);
        reply->set_updatenextindex(-100);  // 论文中：让领导人可以及时更新自己
        // DPrintf("[func-AppendEntries-rf{%d}] 拒绝了 因为Leader{%d}的term{%v}< rf{%d}.term{%d}\n", m_me, args->leaderid(), args->term(), m_me, m_currentTerm);
        LOG << "[func-AppendEntries-rf{" << m_me << "}] 拒绝了 因为Leader{" << args->leaderid() << "}的term{" << args->term() << "}< rf{" << m_me << "}.term{" << m_currentTerm << "}";
        return;  // 注意从过期的领导人收到消息不要重设定时器
    }

    // 由于这个局部变量创建在锁之后，因此执行persist的时候应该也是拿到锁的.
    DEFER{ persist(); };
    if (args->term() > m_currentTerm) {
        // 三变 ,防止遗漏，无论什么时候都是三变
        m_status = Follower;
        m_currentTerm = args->term();
        m_votedFor = -1;  // 这里设置成-1有意义，如果突然宕机然后上线理论上是可以投票的
    }

    myAssert(args->term() == m_currentTerm, format("assert {args.Term == rf.currentTerm} fail"));
    m_status = Follower;  // 这里是有必要的，因为如果candidate收到同一个term的leader的AE，需要变成follower
    // term相等
    m_lastResetElectionTime = now();
    //  DPrintf("[	AppendEntries-func-rf(%v)		] 重置了选举超时定时器\n", rf.me);

    // 不能无脑的从prevlogIndex开始阶段日志，因为rpc可能会延迟，导致发过来的log是很久之前的

    // 那么就比较日志，日志有3种情况
    if (args->prevlogindex() > getLastLogIndex()) {
        reply->set_success(false);
        reply->set_term(m_currentTerm);
        reply->set_updatenextindex(getLastLogIndex() + 1);
        //  DPrintf("[func-AppendEntries-rf{%v}] 拒绝了节点{%v}，因为日志太新,args.PrevLogIndex{%v} >
        //  lastLogIndex{%v}，返回值：{%v}\n", rf.me, args.LeaderId, args.PrevLogIndex, rf.getLastLogIndex(), reply)
        return;
    }
    else if (args->prevlogindex() < m_lastSnapshotIncludeIndex) {
        // 如果prevlogIndex还没有更上快照
        reply->set_success(false);
        reply->set_term(m_currentTerm);
        reply->set_updatenextindex(m_lastSnapshotIncludeIndex + 1);  // todo 如果想直接弄到最新好像不对，因为是从后慢慢往前匹配的，这里不匹配说明后面的都不匹配
        //  DPrintf("[func-AppendEntries-rf{%v}] 拒绝了节点{%v}，因为log太老，返回值：{%v}\n", rf.me, args.LeaderId, reply)
        //  return
    }
    // 本机日志有那么长，冲突(same index,different term),截断日志
    // 注意：这里目前当args.PrevLogIndex == rf.lastSnapshotIncludeIndex与不等的时候要分开考虑，可以看看能不能优化这块
    if (matchLog(args->prevlogindex(), args->prevlogterm())) {
        //	todo：	整理logs，不能直接截断，必须一个一个检查，因为发送来的log可能是之前的，直接截断可能导致“取回”已经在follower日志中的条目
        // 那意思是不是可能会有一段发来的AE中的logs中前半是匹配的，后半是不匹配的，这种应该：1.follower如何处理？ 2.如何给leader回复 3. leader如何处理

        //日志匹配，那么就复制日志
        for (int i = 0; i < args->entries_size(); i++) {
            auto log = args->entries(i);
            if (log.logindex() > getLastLogIndex()) {
                // 超过就直接添加日志
                m_logs.push_back(log);
            }
            else {
                //没超过就比较是否匹配，不匹配再更新，而不是直接截断
                // todo ： 这里可以改进为比较对应logIndex位置的term是否相等，term相等就代表匹配
                //  todo：这个地方放出来会出问题,按理说index相同，term相同，log也应该相同才对
                // rf.logs[entry.Index-firstIndex].Term ?= entry.Term

                if (m_logs[getSlicesIndexFromLogIndex(log.logindex())].logterm() == log.logterm() &&
                    m_logs[getSlicesIndexFromLogIndex(log.logindex())].command() != log.command()) {
                    //相同位置的log ，其logTerm相等，但是命令却不相同，不符合raft的前向匹配，异常了！
                    myAssert(false, format("[func-AppendEntries-rf{%d}] 两节点logIndex{%d}和term{%d}相同，但是其command{%d:%d}   "
                        " {%d:%d}却不同！！\n",
                        m_me, log.logindex(), log.logterm(), m_me,
                        m_logs[getSlicesIndexFromLogIndex(log.logindex())].command(), args->leaderid(),
                        log.command()));
                }
                if (m_logs[getSlicesIndexFromLogIndex(log.logindex())].logterm() != log.logterm()) {
                    //不匹配就更新
                    m_logs[getSlicesIndexFromLogIndex(log.logindex())] = log;
                }
            }
        }

        // 错误写法like：  rf.shrinkLogsToIndex(args.PrevLogIndex)
        // rf.logs = append(rf.logs, args.Entries...)
        // 因为可能会收到过期的log！！！ 因此这里是大于等于
        myAssert(
            getLastLogIndex() >= args->prevlogindex() + args->entries_size(),
            format("[func-AppendEntries1-rf{%d}]rf.getLastLogIndex(){%d} != args.PrevLogIndex{%d}+len(args.Entries){%d}",
                m_me, getLastLogIndex(), args->prevlogindex(), args->entries_size()));
        // if len(args.Entries) > 0 {
        //	fmt.Printf("[func-AppendEntries  rf:{%v}] ] : args.term:%v, rf.term:%v  ,rf.logs的长度：%v\n", rf.me, args.Term,
        // rf.currentTerm, len(rf.logs))
        // }
        if (args->leadercommit() > m_commitIndex) {
            // 这个地方不能无脑跟上getLastLogIndex()，因为可能存在args->leadercommit()落后于 getLastLogIndex()的情况
            m_commitIndex = std::min(args->leadercommit(), getLastLogIndex());
        }

        // 领导会一次发送完所有的日志
        myAssert(getLastLogIndex() >= m_commitIndex,
            format("[func-AppendEntries1-rf{%d}]  rf.getLastLogIndex{%d} < rf.commitIndex{%d}", m_me,
                getLastLogIndex(), m_commitIndex));
        reply->set_success(true);
        reply->set_term(m_currentTerm);

        //        DPrintf("[func-AppendEntries-rf{%v}] 接收了来自节点{%v}的log，当前lastLogIndex{%v}，返回值：{%v}\n",
        //        rf.me,
        //                args.LeaderId, rf.getLastLogIndex(), reply)

        return;
    }
    else {
        // 优化
        // PrevLogIndex 长度合适，但是不匹配，因此往前寻找 矛盾的term的第一个元素
        // 为什么该term的日志都是矛盾的呢？也不一定都是矛盾的，只是这么优化减少rpc而已
        // ？什么时候term会矛盾呢？很多情况，比如leader接收了日志之后马上就崩溃等等
        reply->set_updatenextindex(args->prevlogindex());

        for (int index = args->prevlogindex(); index >= m_lastSnapshotIncludeIndex; --index) {
            if (getLogTermFromLogIndex(index) != getLogTermFromLogIndex(args->prevlogindex())) {
                reply->set_updatenextindex(index + 1);
                break;
            }
        }
        reply->set_success(false);
        reply->set_term(m_currentTerm);
        // 对UpdateNextIndex待优化  todo  找到符合的term的最后一个
        //        DPrintf("[func-AppendEntries-rf{%v}]
        //        拒绝了节点{%v}，因为prevLodIndex{%v}的args.term{%v}不匹配当前节点的logterm{%v}，返回值：{%v}\n",
        //                rf.me, args.LeaderId, args.PrevLogIndex, args.PrevLogTerm,
        //                rf.logs[rf.getSlicesIndexFromLogIndex(args.PrevLogIndex)].LogTerm, reply)
        //        DPrintf("[func-AppendEntries-rf{%v}] 返回值: reply.UpdateNextIndex从{%v}优化到{%v}，优化了{%v}\n", rf.me,
        //                args.PrevLogIndex, reply.UpdateNextIndex, args.PrevLogIndex - reply.UpdateNextIndex) //
        //                很多都是优化了0
        return;
    }

    // fmt.Printf("[func-AppendEntries,rf{%v}]:len(rf.logs):%v, rf.commitIndex:%v\n", rf.me, len(rf.logs), rf.commitIndex)
}

void Raft::applierTicker() {
    while (true) {
        // m_mtx.lock();
        if (m_status == Leader) {
            // LOG << "[Raft::applierTicker() - raft{" << m_me << "}]  m_lastApplied{" << m_lastApplied << "}   m_commitIndex{" << m_commitIndex << "}";
        }
        auto applyMsgs = getApplyLogs();
        // m_mtx.unlock();
        //使用匿名函数是因为传递管道的时候不用拿锁
        // todo:好像必须拿锁，因为不拿锁的话如果调用多次applyLog函数，可能会导致应用的顺序不一样
        if (!applyMsgs.empty()) {
            // LOG << "[func- Raft::applierTicker()-raft{" << m_me << "}] 向kvserver报告的applyMsgs长度为：{" << applyMsgs.size() << "}";
        }
        for (auto& msg : applyMsgs) {
            applyChan->Push(msg);
        }

        std::this_thread::sleep_for(std::chrono::microseconds(ApplyInterval * 2));

    }
}

bool Raft::CondInstallSnapshot(int lastIncludedTerm, int lastIncludedIndex, std::string snapshot) {
    return true;
    //// Your code here (2D).
    // rf.mu.Lock()
    // defer rf.mu.Unlock()
    // DPrintf("{Node %v} service calls CondInstallSnapshot with lastIncludedTerm %v and lastIncludedIndex {%v} to check
    // whether snapshot is still valid in term %v", rf.me, lastIncludedTerm, lastIncludedIndex, rf.currentTerm)
    //// outdated snapshot
    // if lastIncludedIndex <= rf.commitIndex {
    //	return false
    // }
    //
    // lastLogIndex, _ := rf.getLastLogIndexAndTerm()
    // if lastIncludedIndex > lastLogIndex {
    //	rf.logs = make([]LogEntry, 0)
    // } else {
    //	rf.logs = rf.logs[rf.getSlicesIndexFromLogIndex(lastIncludedIndex)+1:]
    // }
    //// update dummy entry with lastIncludedTerm and lastIncludedIndex
    // rf.lastApplied, rf.commitIndex = lastIncludedIndex, lastIncludedIndex
    //
    // rf.persister.Save(rf.persistData(), snapshot)
    // return true
}

void Raft::doElection() {
    std::lock_guard<std::mutex> g(m_mtx);

    // if (m_status == Leader) {
    //     // fmt.Printf("[       ticker-func-rf(%v)              ] is a Leader,wait the  lock\n", rf.me)
    // }

    if (m_status != Leader) {
        LOG << "[ticker-func-rf{" << m_me << "}]  选举定时器到期且不是leader，开始选举 ";
        //当选举的时候定时器超时就必须重新选举，不然没有选票就会一直卡主
        //重竞选超时，term也会增加的
        m_status = Candidate;
        ///开始新一轮的选举
        m_currentTerm += 1;
        m_votedFor = m_me;  //即是自己给自己投，也避免candidate给同辈的candidate投
        persist(); //持久化?????????????????????????????????????????????????????????
        std::shared_ptr<int> votedNum = std::make_shared<int>(1);  // 使用 make_shared 函数初始化 !! 亮点
        //	重新设置定时器
        m_lastResetElectionTime = now();
        //	发布RequestVote RPC
        for (int i = 0; i < m_peers.size(); i++) {
            if (i == m_me) {
                continue;
            }
            int lastLogIndex = -1, lastLogTerm = -1;
            getLastLogIndexAndTerm(&lastLogIndex, &lastLogTerm);  //获取最后一个log的term和下标

            std::shared_ptr<raftRpcProctoc::RequestVoteArgs> requestVoteArgs = std::make_shared<raftRpcProctoc::RequestVoteArgs>();
            requestVoteArgs->set_term(m_currentTerm);
            requestVoteArgs->set_candidateid(m_me);
            requestVoteArgs->set_lastlogindex(lastLogIndex);
            requestVoteArgs->set_lastlogterm(lastLogTerm);
            auto requestVoteReply = std::make_shared<raftRpcProctoc::RequestVoteReply>();

            //使用匿名函数执行避免其拿到锁

            // std::thread t(&Raft::sendRequestVote, this, i, requestVoteArgs, requestVoteReply, votedNum);  // 创建新线程并执行b函数，并传递参数
            // t.detach();
            // 使用线程池执行成员函数
            // m_threadPool->enqueue(&Raft::sendRequestVote, this, i, requestVoteArgs, requestVoteReply, votedNum);
            m_threadPool->enqueue([this, i, requestVoteArgs, requestVoteReply, votedNum]() {
                this->sendRequestVote(i, requestVoteArgs, requestVoteReply, votedNum);
            });
        }
    }
}

void Raft::doHeartBeat() {
    // std::lock_guard<std::mutex> g(m_mtx);

    if (m_status == Leader) {
        // LOG << "[func-Raft::doHeartBeat()-Leader: {" << m_me << "}] Leader的心跳定时器触发了且拿到mutex，开始发送AE";
        auto appendNums = std::make_shared<int>(1);  //正确返回的节点的数量

        //对Follower（除了自己外的所有节点发送AE）
        // todo 这里肯定是要修改的，最好使用一个单独的goruntime来负责管理发送log，因为后面的log发送涉及优化之类的
        //最少要单独写一个函数来管理，而不是在这一坨
        for (int i = 0; i < m_peers.size(); i++) {
            if (i == m_me) {
                continue;
            }
            // LOG << "[func-Raft::doHeartBeat()-Leader: {" << m_me << "}] Leader的心跳定时器触发了 index:{" << i << "}";
            myAssert(m_nextIndex[i] >= 1, format("rf.nextIndex[%d] = {%d}", i, m_nextIndex[i]));

            //日志压缩加入后要判断是发送快照还是发送AE
            if (m_nextIndex[i] <= m_lastSnapshotIncludeIndex) {
                //该发送的日志已经被做成快照，必须发送快照了
                std::thread t(&Raft::leaderSendSnapShot, this, i);  // 创建新线程并执行b函数，并传递参数
                t.detach();
                continue;
            }

            //发送心跳，构造发送值
            int preLogIndex = -1;
            int PrevLogTerm = -1;
            getPrevLogInfo(i, &preLogIndex, &PrevLogTerm); //获取本次发送的一系列日志的上一条日志的信息，以判断是否匹配
            std::shared_ptr<raftRpcProctoc::AppendEntriesArgs> appendEntriesArgs = std::make_shared<raftRpcProctoc::AppendEntriesArgs>();
            appendEntriesArgs->set_term(m_currentTerm);
            appendEntriesArgs->set_leaderid(m_me);
            appendEntriesArgs->set_prevlogindex(preLogIndex);
            appendEntriesArgs->set_prevlogterm(PrevLogTerm);
            appendEntriesArgs->clear_entries();
            appendEntriesArgs->set_leadercommit(m_commitIndex);
            if (preLogIndex != m_lastSnapshotIncludeIndex) {
                for (int j = getSlicesIndexFromLogIndex(preLogIndex) + 1; j < m_logs.size(); ++j) {
                    // appendEntriesArgs->add_entries()返回的是一个指针，指向新创建的LogEntry
                    raftRpcProctoc::LogEntry* sendEntryPtr = appendEntriesArgs->add_entries();
                    *sendEntryPtr = m_logs[j];
                }
            }
            else {
                for (const auto& item : m_logs) {
                    raftRpcProctoc::LogEntry* sendEntryPtr = appendEntriesArgs->add_entries();
                    *sendEntryPtr = item;
                }
            }
            int lastLogIndex = getLastLogIndex();
            // leader对每个节点发送的日志长短不一，但是都保证从prevIndex发送直到最后
            myAssert(appendEntriesArgs->prevlogindex() + appendEntriesArgs->entries_size() == lastLogIndex,
                format("appendEntriesArgs.PrevLogIndex{%d}+len(appendEntriesArgs.Entries){%d} != lastLogIndex{%d}",
                    appendEntriesArgs->prevlogindex(), appendEntriesArgs->entries_size(), lastLogIndex));
            //构造返回值
            const std::shared_ptr<raftRpcProctoc::AppendEntriesReply> appendEntriesReply = std::make_shared<raftRpcProctoc::AppendEntriesReply>();
            appendEntriesReply->set_appstate(Disconnected);

            // std::thread t(&Raft::sendAppendEntries, this, i, appendEntriesArgs, appendEntriesReply, appendNums);  // 创建新线程并执行b函数，并传递参数
            // t.detach();
            // m_threadPool->enqueue(&Raft::sendAppendEntries, this, i, appendEntriesArgs, appendEntriesReply, appendNums);

            m_threadPool->enqueue([this, i, appendEntriesArgs, appendEntriesReply, appendNums]() {
                this->sendAppendEntries(i, appendEntriesArgs, appendEntriesReply, appendNums);
            });
            
        }
        m_lastResetHearBeatTime = now();  // leader发送心跳，就不是随机时间了
    }
}

void Raft::electionTimeOutTicker() {
    // Check if a Leader election should be started.
    while (true) {
        /**
         * 如果不睡眠，那么对于leader，这个函数会一直空转，浪费cpu。且加入协程之后，空转会导致其他协程无法运行，对于时间敏感的AE，会导致心跳无法正常发送导致异常
         */
        while (m_status == Leader) {
            usleep(HeartBeatTimeout);  //定时时间没有严谨设置，因为HeartBeatTimeout比选举超时一般小一个数量级，因此就设置为HeartBeatTimeout了
        }
        std::chrono::duration<signed long int, std::ratio<1, 1000000000>> suitableSleepTime{};
        std::chrono::system_clock::time_point wakeTime{};
        {
            m_mtx.lock();
            wakeTime = now(); //睡眠前记录时间
            suitableSleepTime = getRandomizedElectionTimeout() + m_lastResetElectionTime - wakeTime;
            m_mtx.unlock();
        }

        if (std::chrono::duration<double, std::milli>(suitableSleepTime).count() > 1) {
            usleep(std::chrono::duration_cast<std::chrono::microseconds>(suitableSleepTime).count());
        }

        if (std::chrono::duration<double, std::milli>(m_lastResetElectionTime - wakeTime).count() > 0) {
            //说明睡眠的这段时间有重置定时器，那么就没有超时，再次睡眠
            continue;
        }
        doElection();
    }
}

std::vector<ApplyMsg> Raft::getApplyLogs() {
    std::vector<ApplyMsg> applyMsgs;
    myAssert(m_commitIndex <= getLastLogIndex(), format("[func-getApplyLogs-rf{%d}] commitIndex{%d} >getLastLogIndex{%d}",
        m_me, m_commitIndex, getLastLogIndex()));

    while (m_lastApplied < m_commitIndex) {
        m_lastApplied++;
        myAssert(m_logs[getSlicesIndexFromLogIndex(m_lastApplied)].logindex() == m_lastApplied,
            format("rf.logs[rf.getSlicesIndexFromLogIndex(rf.lastApplied)].LogIndex{%d} != rf.lastApplied{%d} ",
                m_logs[getSlicesIndexFromLogIndex(m_lastApplied)].logindex(), m_lastApplied));
        ApplyMsg applyMsg;
        applyMsg.CommandValid = true;
        applyMsg.Command = m_logs[getSlicesIndexFromLogIndex(m_lastApplied)].command();
        applyMsg.CommandIndex = m_lastApplied;
        applyMsgs.emplace_back(applyMsg);
        //        DPrintf("[	applyLog func-rf{%v}	] apply Log,logIndex:%v  ，logTerm：{%v},command：{%v}\n",
        //        rf.me, rf.lastApplied, rf.logs[rf.getSlicesIndexFromLogIndex(rf.lastApplied)].LogTerm,
        //        rf.logs[rf.getSlicesIndexFromLogIndex(rf.lastApplied)].Command)
    }
    return applyMsgs;
}

// 获取新命令应该分配的Index
int Raft::getNewCommandIndex() {
    //	如果len(logs)==0,就为快照的index+1，否则为log最后一个日志+1
    auto lastLogIndex = getLastLogIndex();
    return lastLogIndex + 1;
}

// getPrevLogInfo
// leader调用，传入：服务器index，传出：发送的AE的preLogIndex和PrevLogTerm
void Raft::getPrevLogInfo(int server, int* preIndex, int* preTerm) {
    // logs长度为0返回0,0，不是0就根据nextIndex数组的数值返回
    if (m_nextIndex[server] == m_lastSnapshotIncludeIndex + 1) {
        //要发送的日志是第一个日志，因此直接返回m_lastSnapshotIncludeIndex和m_lastSnapshotIncludeTerm
        *preIndex = m_lastSnapshotIncludeIndex;
        *preTerm = m_lastSnapshotIncludeTerm;
        return;
    }
    auto nextIndex = m_nextIndex[server];
    *preIndex = nextIndex - 1;
    *preTerm = m_logs[getSlicesIndexFromLogIndex(*preIndex)].logterm();
}

// GetState return currentTerm and whether this server
// believes it is the Leader.
void Raft::GetState(int* term, bool* isLeader) {
    m_mtx.lock();
    DEFER{
        // todo 暂时不清楚会不会导致死锁
        m_mtx.unlock();
    };

    // Your code here (2A).
    *term = m_currentTerm;
    *isLeader = (m_status == Leader);
}

void Raft::InstallSnapshot(const raftRpcProctoc::InstallSnapshotRequest* args,
    raftRpcProctoc::InstallSnapshotResponse* reply) {
    m_mtx.lock();
    DEFER{ m_mtx.unlock(); };
    if (args->term() < m_currentTerm) {
        reply->set_term(m_currentTerm);
        return;
    }
    if (args->term() > m_currentTerm) {
        //后面两种情况都要接收日志
        m_currentTerm = args->term();
        m_votedFor = -1;
        m_status = Follower;
        persist();
    }
    m_status = Follower;
    m_lastResetElectionTime = now();
    // outdated snapshot
    if (args->lastsnapshotincludeindex() <= m_lastSnapshotIncludeIndex) {
        return;
    }
    //截断日志，修改commitIndex和lastApplied
    //截断日志包括：日志长了，截断一部分，日志短了，全部清空，其实两个是一种情况
    //但是由于现在getSlicesIndexFromLogIndex的实现，不能传入不存在logIndex，否则会panic
    auto lastLogIndex = getLastLogIndex();

    if (lastLogIndex > args->lastsnapshotincludeindex()) {
        m_logs.erase(m_logs.begin(), m_logs.begin() + getSlicesIndexFromLogIndex(args->lastsnapshotincludeindex()) + 1);
    }
    else {
        m_logs.clear();
    }
    m_commitIndex = std::max(m_commitIndex, args->lastsnapshotincludeindex());
    m_lastApplied = std::max(m_lastApplied, args->lastsnapshotincludeindex());
    m_lastSnapshotIncludeIndex = args->lastsnapshotincludeindex();
    m_lastSnapshotIncludeTerm = args->lastsnapshotincludeterm();

    reply->set_term(m_currentTerm);
    ApplyMsg msg;
    msg.SnapshotValid = true;
    msg.Snapshot = args->data();
    msg.SnapshotTerm = args->lastsnapshotincludeterm();
    msg.SnapshotIndex = args->lastsnapshotincludeindex();

    // std::thread t(&Raft::pushMsgToKvServer, this, msg);  // 创建新线程并执行b函数，并传递参数
    // t.detach();
    applyChan->Push(msg);

    //看下这里能不能再优化
    //持久化
    m_persister->Save(persistData(), args->data());
}

// void Raft::pushMsgToKvServer(ApplyMsg msg) { applyChan->Push(msg); }

// void Raft::leaderHearBeatTicker() {
//     while (true) {
//         //不是leader的话就没有必要进行后续操作，况且还要拿锁，很影响性能，目前是睡眠，后面再优化优化
//         while (m_status != Leader) {
//             // usleep(1000 * HeartBeatTimeout);
//             std::this_thread::sleep_for(std::chrono::microseconds(200 * HeartBeatTimeout));
//         }
//         static std::atomic<int32_t> atomicCount = 0;

//         std::chrono::duration<signed long int, std::ratio<1, 1000000000>> suitableSleepTime{};
//         std::chrono::system_clock::time_point wakeTime{};
//         {
//             std::lock_guard<std::mutex> lock(m_mtx);
//             wakeTime = now();
//             // suitableSleepTime = std::chrono::milliseconds(HeartBeatTimeout) + m_lastResetHearBeatTime - wakeTime;
//             suitableSleepTime = std::chrono::microseconds(1000 * HeartBeatTimeout) + m_lastResetHearBeatTime - wakeTime;
//         }

//         if (std::chrono::duration<double, std::milli>(suitableSleepTime).count() > 1) {
//             // 获取当前时间点
//             auto start = std::chrono::steady_clock::now();

//             // usleep(std::chrono::duration_cast<std::chrono::microseconds>(suitableSleepTime).count());
//             std::this_thread::sleep_for(std::chrono::nanoseconds(suitableSleepTime.count()));

//             // 获取函数运行结束后的时间点
//             auto end = std::chrono::steady_clock::now();

//             // 计算时间差并输出结果（单位为毫秒）
//             std::chrono::duration<double, std::milli> duration = end - start;

//             // LOG<< atomicCount<<"[func-leaderHearBeatTicker();函数实际睡眠时间为: " << duration.count() << " ms";
//             ++atomicCount;
//         }

//         if (std::chrono::duration<double, std::milli>(m_lastResetHearBeatTime - wakeTime).count() > 0) {
//             //睡眠的这段时间有重置定时器，没有超时，再次睡眠
//             continue;
//         }
//         // LOG << "[func-Raft::doHeartBeat()-Leader: {" << m_me << "}] Leader的心跳定时器触发了";
//         doHeartBeat();
//     }
// }

void Raft::leaderHearBeatTicker() {
    while (true) {
        // 如果不是 Leader，等待后重新检查状态
        while (m_status != Leader) {
            std::this_thread::sleep_for(std::chrono::microseconds(200 * HeartBeatTimeout));
        }

        // 每次进入 Leader 状态时，重置心跳时间
        {
            std::lock_guard<std::mutex> lock(m_mtx);
            m_lastResetHearBeatTime = now();
        }

        while (m_status == Leader) {
            std::unique_lock<std::mutex> lock(m_mtx);

            // 下一次心跳触发的时间点
            auto nextHeartbeatTime = m_lastResetHearBeatTime + std::chrono::microseconds(1000 * HeartBeatTimeout);
            bool newLogAvailable = false;
            // 等待直到超时或有新日志/状态变化
            bool notified = m_newLogCond.wait_until(lock, nextHeartbeatTime, [this]() {
                return m_status != Leader || m_newLogAvailable;
                });
            // 检查是否是因为状态变化而唤醒
            if (m_status != Leader) {
                lock.unlock(); // 释放锁，允许其他线程操作！！！！！！！
                break;
            }
            // 如果是因为新日志而唤醒，重置标志
            if (m_newLogAvailable) {
                LOG << "新日志到达，距离上次心跳时间为：" << std::chrono::duration_cast<std::chrono::microseconds>(now() - m_lastResetHearBeatTime).count() << " us";
                newLogAvailable = true;
                m_newLogAvailable = false;
            }
            lock.unlock(); // 释放锁，允许其他线程操作！！！！！！！

            // 检查当前时间是否超过预期心跳时间
            auto nowTime = now();
            if (nowTime >= nextHeartbeatTime || newLogAvailable) {
                doHeartBeat();
                {
                    std::lock_guard<std::mutex> lock(m_mtx);
                    m_lastResetHearBeatTime = nowTime; // 更新心跳时间
                }
            }
        }
    }
}

void Raft::leaderSendSnapShot(int server) {
    m_mtx.lock();
    raftRpcProctoc::InstallSnapshotRequest args;
    args.set_leaderid(m_me);
    args.set_term(m_currentTerm);
    args.set_lastsnapshotincludeindex(m_lastSnapshotIncludeIndex);
    args.set_lastsnapshotincludeterm(m_lastSnapshotIncludeTerm);
    args.set_data(m_persister->ReadSnapshot());

    raftRpcProctoc::InstallSnapshotResponse reply;
    m_mtx.unlock();
    // 通过rpc客户端发送快照
    bool ok = m_peers[server]->InstallSnapshot(&args, &reply);

    std::lock_guard<std::mutex> lg(m_mtx);
    // m_mtx.lock();
    // DEFER{ m_mtx.unlock(); };
    if (!ok) {
        return;
    }
    if (m_status != Leader || m_currentTerm != args.term()) {
        return;  //中间释放过锁，可能状态已经改变了
    }
    //	无论什么时候都要判断term
    if (reply.term() > m_currentTerm) {
        //三变
        m_currentTerm = reply.term();
        m_votedFor = -1;
        m_status = Follower;
        persist();
        m_lastResetElectionTime = now();
        return;
    }
    m_matchIndex[server] = args.lastsnapshotincludeindex();
    m_nextIndex[server] = m_matchIndex[server] + 1;
}

void Raft::leaderUpdateCommitIndex() {
    m_commitIndex = m_lastSnapshotIncludeIndex;
    // for index := rf.commitIndex+1;index < len(rf.log);index++ {
    // for index := rf.getLastIndex();index>=rf.commitIndex+1;index--{
    for (int index = getLastLogIndex(); index >= m_lastSnapshotIncludeIndex + 1; index--) {
        int sum = 0;
        for (int i = 0; i < m_peers.size(); i++) {
            if (i == m_me) {
                sum += 1;
                continue;
            }
            if (m_matchIndex[i] >= index) {
                sum += 1;
            }
        }

        //        !!!只有当前term有新提交的，才会更新commitIndex！！！！
        // log.Printf("lastSSP:%d, index: %d, commitIndex: %d, lastIndex: %d",rf.lastSSPointIndex, index, rf.commitIndex,
        // rf.getLastIndex())
        if (sum >= m_peers.size() / 2 + 1 && getLogTermFromLogIndex(index) == m_currentTerm) {
            m_commitIndex = index;
            break;
        }
    }
    //    DPrintf("[func-leaderUpdateCommitIndex()-rf{%v}] Leader %d(term%d) commitIndex
    //    %d",rf.me,rf.me,rf.currentTerm,rf.commitIndex)
}

//进来前要保证logIndex是存在的，即≥rf.lastSnapshotIncludeIndex	，而且小于等于rf.getLastLogIndex()
bool Raft::matchLog(int logIndex, int logTerm) {
    myAssert(logIndex >= m_lastSnapshotIncludeIndex && logIndex <= getLastLogIndex(),
        format("不满足：logIndex{%d}>=rf.lastSnapshotIncludeIndex{%d}&&logIndex{%d}<=rf.getLastLogIndex{%d}",
            logIndex, m_lastSnapshotIncludeIndex, logIndex, getLastLogIndex()));
    return logTerm == getLogTermFromLogIndex(logIndex);
    // if logIndex == rf.lastSnapshotIncludeIndex {
    // 	return logTerm == rf.lastSnapshotIncludeTerm
    // } else {
    // 	return logTerm == rf.logs[rf.getSlicesIndexFromLogIndex(logIndex)].LogTerm
    // }
}

void Raft::persist() {
    // Your code here (2C).
    return;
    auto data = persistData();
    m_persister->SaveRaftState(data);
}

void Raft::RequestVote(const raftRpcProctoc::RequestVoteArgs* args, raftRpcProctoc::RequestVoteReply* reply) {
    std::lock_guard<std::mutex> lg(m_mtx);

    // Your code here (2A, 2B).
    DEFER{
        //应该先持久化，再撤销lock
        persist();
    };
    //对args的term的三种情况分别进行处理，大于小于等于自己的term都是不同的处理
    // reason: 出现网络分区，该竞选者已经OutOfDate(过时）
    if (args->term() < m_currentTerm) {
        reply->set_term(m_currentTerm);
        reply->set_votestate(Expire);
        reply->set_votegranted(false);
        return;
    }

    // fig2:右下角，如果任何时候rpc请求或者响应的term大于自己的term，更新term，并变成follower
    if (args->term() > m_currentTerm) {
        m_status = Follower;
        m_currentTerm = args->term();
        m_votedFor = -1;

        //	重置定时器：收到leader的ae，开始选举，投出票
        // 这时候更新了term之后，votedFor也要置为-1
    }

    myAssert(args->term() == m_currentTerm, format("[func--rf{%d}] 前面校验过args.Term==rf.currentTerm，这里却不等", m_me));
    //	现在节点任期都是相同的(任期小的也已经更新到新的args的term了)，还需要检查log的term和index是不是匹配的了
    int lastLogTerm = getLastLogTerm();
    // 只有没投票，且candidate的日志的新的程度 ≥ 接受者的日志新的程度 才会授票
    if (!UpToDate(args->lastlogindex(), args->lastlogterm())) {
        // args.LastLogTerm < lastLogTerm || (args.LastLogTerm == lastLogTerm && args.LastLogIndex < lastLogIndex) {
        // 日志太旧了
        // if (args->lastlogterm() < lastLogTerm) {
        //     //                    DPrintf("[	    func-RequestVote-rf(%v)		] : refuse voted rf[%v] ,because
        //     //                    candidate_lastlog_term{%v} < lastlog_term{%v}\n", rf.me, args.CandidateId, args.LastLogTerm,
        //     //                    lastLogTerm)
        // }
        // else {
        //     //            DPrintf("[	    func-RequestVote-rf(%v)		] : refuse voted rf[%v] ,because
        //     //            candidate_log_index{%v} < log_index{%v}\n", rf.me, args.CandidateId, args.LastLogIndex,
        //     //            rf.getLastLogIndex())
        // }

        // 日志太旧了
        reply->set_term(m_currentTerm);
        reply->set_votestate(Voted);
        reply->set_votegranted(false);

        return;
    }
    // todo ： 啥时候会出现rf.votedFor == args.CandidateId ，就算candidate选举超时再选举，其term也是不一样的呀？
    // 当因为网络质量不好导致的请求丢失重发就有可能！！！！
    // 因此需要避免重复投票
    if (m_votedFor != -1 && m_votedFor != args->candidateid()) {
        //        DPrintf("[	    func-RequestVote-rf(%v)		] : refuse voted rf[%v] ,because has voted\n",
        //        rf.me, args.CandidateId)
        reply->set_term(m_currentTerm);
        reply->set_votestate(Voted);
        reply->set_votegranted(false);

        return;
    }
    else {
        m_votedFor = args->candidateid();
        m_lastResetElectionTime = now();  //认为必须要在投出票的时候才重置定时器，
        //        DPrintf("[	    func-RequestVote-rf(%v)		] : voted rf[%v]\n", rf.me, rf.votedFor)
        reply->set_term(m_currentTerm);
        reply->set_votestate(Normal);
        reply->set_votegranted(true);

        return;
    }
}

bool Raft::UpToDate(int index, int term) {
    // lastEntry := rf.log[len(rf.log)-1]

    int lastIndex = -1;
    int lastTerm = -1;
    getLastLogIndexAndTerm(&lastIndex, &lastTerm);
    return term > lastTerm || (term == lastTerm && index >= lastIndex);
}

void Raft::getLastLogIndexAndTerm(int* lastLogIndex, int* lastLogTerm) {
    if (m_logs.empty()) {
        *lastLogIndex = m_lastSnapshotIncludeIndex;
        *lastLogTerm = m_lastSnapshotIncludeTerm;
        return;
    }
    else {
        *lastLogIndex = m_logs[m_logs.size() - 1].logindex();
        *lastLogTerm = m_logs[m_logs.size() - 1].logterm();
        return;
    }
}
/**
 *
 * @return 最新的log的logindex，即log的逻辑index。区别于log在m_logs中的物理index
 * 可见：getLastLogIndexAndTerm()
 */
int Raft::getLastLogIndex() {
    int lastLogIndex = -1;
    int _ = -1;
    getLastLogIndexAndTerm(&lastLogIndex, &_);
    return lastLogIndex;
}

int Raft::getLastLogTerm() {
    int _ = -1;
    int lastLogTerm = -1;
    getLastLogIndexAndTerm(&_, &lastLogTerm);
    return lastLogTerm;
}

/**
 *
 * @param logIndex log的逻辑index。注意区别于m_logs的物理index
 * @return
 */
int Raft::getLogTermFromLogIndex(int logIndex) {
    myAssert(logIndex >= m_lastSnapshotIncludeIndex,
        format("[func-getSlicesIndexFromLogIndex-rf{%d}]  index{%d} < rf.lastSnapshotIncludeIndex{%d}", m_me,
            logIndex, m_lastSnapshotIncludeIndex));

    int lastLogIndex = getLastLogIndex();

    myAssert(logIndex <= lastLogIndex, format("[func-getSlicesIndexFromLogIndex-rf{%d}]  logIndex{%d} > lastLogIndex{%d}",
        m_me, logIndex, lastLogIndex));

    if (logIndex == m_lastSnapshotIncludeIndex) {
        return m_lastSnapshotIncludeTerm;
    }
    else {
        return m_logs[getSlicesIndexFromLogIndex(logIndex)].logterm();
    }
}

int Raft::GetRaftStateSize() { return m_persister->RaftStateSize(); }

// 找到index对应的真实下标位置！！！
// 限制，输入的logIndex必须保存在当前的logs里面（不包含snapshot）
int Raft::getSlicesIndexFromLogIndex(int logIndex) {
    myAssert(logIndex > m_lastSnapshotIncludeIndex,
        format("[func-getSlicesIndexFromLogIndex-rf{%d}]  index{%d} <= rf.lastSnapshotIncludeIndex{%d}", m_me,
            logIndex, m_lastSnapshotIncludeIndex));
    int lastLogIndex = getLastLogIndex();
    myAssert(logIndex <= lastLogIndex, format("[func-getSlicesIndexFromLogIndex-rf{%d}]  logIndex{%d} > lastLogIndex{%d}",
        m_me, logIndex, lastLogIndex));
    int SliceIndex = logIndex - m_lastSnapshotIncludeIndex - 1;
    return SliceIndex;
}

bool Raft::sendRequestVote(int server, std::shared_ptr<raftRpcProctoc::RequestVoteArgs> args,
    std::shared_ptr<raftRpcProctoc::RequestVoteReply> reply, std::shared_ptr<int> votedNum) {
    //这个ok是网络是否正常通信的ok，而不是requestVote rpc是否投票的rpc
    // ok := rf.peers[server].Call("Raft.RequestVote", args, reply)
    // todo
    auto start = now();
    // DPrintf("[func-sendRequestVote rf{%d}] 向server{%d} 发送 RequestVote 开始", m_me, m_currentTerm, getLastLogIndex());
    LOG << "[func-sendRequestVote rf{" << m_me << "}] 向server{" << server << "} 发送 RequestVote 开始";
    bool ok = m_peers[server]->RequestVote(args.get(), reply.get());
    LOG << "[func-sendRequestVote rf{" << m_me << "}] 向server{" << server << "} 发送 RequestVote 完毕，耗时:{" << std::chrono::duration_cast<std::chrono::milliseconds>(now() - start).count() << "} ms";

    if (!ok) {
        //rpc通信失败就立即返回，避免资源消耗
        return ok;  //不知道为什么不加这个的话如果服务器宕机会出现问题的，通不过2B  todo
    }
    // for !ok {
    //
    //	//ok := rf.peers[server].Call("Raft.RequestVote", args, reply)
    //	//if ok {
    //	//	break
    //	//}
    // } //这里是发送出去了，但是不能保证他一定到达
    //对回应进行处理，要记得无论什么时候收到回复就要检查term
    std::lock_guard<std::mutex> lg(m_mtx);
    if (reply->term() > m_currentTerm) {
        //回复的term比自己大，说明自己落后了，那么就更新自己的状态并且退出
        m_status = Follower;  //三变：身份，term，和投票
        m_currentTerm = reply->term();
        m_votedFor = -1;
        persist();  //持久化
        return true;
    }
    else if (reply->term() < m_currentTerm) {
        //回复的term比自己的term小，不应该出现这种情况
        return true;
    }
    myAssert(reply->term() == m_currentTerm, format("assert {reply.Term==rf.currentTerm} fail"));

    // todo：这里没有按博客写
    if (!reply->votegranted()) {
        //这个节点因为某些原因没给自己投票，没啥好说的，结束本函数
        return true;
    }

    //给自己投票了
    *votedNum = *votedNum + 1;
    if (*votedNum >= m_peers.size() / 2 + 1) {
        //变成leader

        //重置voteDNum，如果不重置，那么就会变成leader很多次，是没有必要的！！！
        *votedNum = 0;
        if (m_status == Leader) {
            //如果已经是leader了，那么是就是了，不会进行下一步处理了k
            myAssert(false, format("[func-sendRequestVote-rf{%d}]  term:{%d} 同一个term当两次领导，error", m_me, m_currentTerm));
        }
        //	第一次变成leader，初始化状态和nextIndex、matchIndex
        m_status = Leader;
        // 胜选后立即

        // DPrintf("[func-sendRequestVote rf{%d}] elect success  ,current term:{%d} ,lastLogIndex:{%d}\n", m_me, m_currentTerm, getLastLogIndex());
        LOG << "[func-sendRequestVote rf{" << m_me << "}] elect success  ,current term:{" << m_currentTerm << "} ,lastLogIndex:{" << getLastLogIndex() << "}";

        // 只有leader才需要维护m_nextIndex和m_matchIndex
        int lastLogIndex = getLastLogIndex();
        for (int i = 0; i < m_nextIndex.size(); i++) {
            m_nextIndex[i] = lastLogIndex + 1;  //有效下标从1开始，因此要+1
            m_matchIndex[i] = 0;                //每换一个领导都是从0开始，见fig2
        }
        std::thread t(&Raft::doHeartBeat, this);  //马上向其他节点宣告自己就是leader
        t.detach();

        persist();
    }
    return true;
}

bool Raft::sendAppendEntries(int server, std::shared_ptr<raftRpcProctoc::AppendEntriesArgs> args,
    std::shared_ptr<raftRpcProctoc::AppendEntriesReply> reply,
    std::shared_ptr<int> appendNums) {
    //这个ok是网络是否正常通信的ok，而不是requestVote rpc是否投票的rpc
    // 如果网络不通的话肯定是没有返回的，不用一直重试
    // todo： paper中5.3节第一段末尾提到，如果append失败应该不断的retries ,直到这个log成功的被store
    // DPrintf("[func-Raft::sendAppendEntries-raft{%d}] leader 向节点{%d}发送AE rpc开始 ， args->entries_size():{%d}", m_me,
    //     server, args->entries_size());
    LOG << "[func-Raft::sendAppendEntries-raft{" << m_me << "}] leader 向节点{" << server
        << "}发送AE rpc开始 ， args->entries_size():{" << args->entries_size() << "}";
    bool ok = m_peers[server]->AppendEntries(args.get(), reply.get());

    if (!ok) {
        // DPrintf("[func-Raft::sendAppendEntries-raft{%d}] leader 向节点{%d}发送AE rpc失败", m_me, server);
        LOG << "[func-Raft::sendAppendEntries-raft{" << m_me << "}] leader 向节点{" << server
            << "}发送AE rpc失败";
        return ok;
    }
    // DPrintf("[func-Raft::sendAppendEntries-raft{%d}] leader 向节点{%d}发送AE rpc成功", m_me, server);
    LOG << "[func-Raft::sendAppendEntries-raft{" << m_me << "}] leader 向节点{" << server
        << "}发送AE rpc成功";
    if (reply->appstate() == Disconnected) {
        return ok;
    }
    std::lock_guard<std::mutex> lg1(m_mtx);

    //对reply进行处理
    // 对于rpc通信，无论什么时候都要检查term
    if (reply->term() > m_currentTerm) {
        m_status = Follower;
        m_currentTerm = reply->term();
        m_votedFor = -1;
        return ok;
    }
    else if (reply->term() < m_currentTerm) {
        //正常不会发生
        // DPrintf("[func -sendAppendEntries  rf{%d}]  节点：{%d}的term{%d}<rf{%d}的term{%d}\n", m_me, server, reply->term(),
        //     m_me, m_currentTerm);
        LOG << "[func -sendAppendEntries  rf{" << m_me << "}]  节点：{" << server
            << "}的term{" << reply->term() << "} < rf{" << m_me << "}的term{" << m_currentTerm << "}";
        return ok;
    }

    if (m_status != Leader) {
        //如果不是leader，那么就不要对返回的情况进行处理了
        return ok;
    }

    // term相等

    myAssert(reply->term() == m_currentTerm,
        format("reply.Term{%d} != rf.currentTerm{%d}   ", reply->term(), m_currentTerm));
    if (!reply->success()) {
        // 日志不匹配，正常来说就是index要往前-1，既然能到这里，第一个日志（idnex = 1）发送后肯定是匹配的，因此不用考虑变成负数 
        // 因为真正的环境不会知道是服务器宕机还是发生网络分区了
        if (reply->updatenextindex() != -100) {
            // todo:待总结，就算term匹配，失败的时候nextIndex也不是照单全收的，因为如果发生rpc延迟，leader的term可能从不符合term要求
            //变得符合term要求
            //但是不能直接赋值reply.UpdateNextIndex
            // DPrintf("[func -sendAppendEntries  rf{%d}]  返回的日志term相等，但是不匹配，回缩nextIndex[%d]：{%d}\n", m_me,
            //     server, reply->updatenextindex());
            LOG << "[func -sendAppendEntries  rf{" << m_me << "}]  返回的日志term相等，但是不匹配，回缩nextIndex["
                << server << "]：{" << reply->updatenextindex() << "}";
            // 优化日志匹配，让follower决定到底应该下一次从哪一个开始尝试发送
            m_nextIndex[server] = reply->updatenextindex();  //失败是不更新mathIndex的
        }
        // 如果感觉rf.nextIndex数组是冗余的，看下论文fig2，其实不是冗余的
    }
    else {
        //到这里代表同意接收了本次心跳或者日志
        *appendNums = *appendNums + 1;
        // DPrintf("---------------------------tmp------------------------- 节点{%d}返回true,当前*appendNums{%d}", server, *appendNums);
        LOG << "---------------------------tmp------------------------- 节点{" << server
            << "}返回true,当前*appendNums{" << *appendNums << "}";
        // rf.matchIndex[server] = len(args.Entries) //只要返回一个响应就对其matchIndex应该对其做出反应，
        //但是这么修改是有问题的，如果对某个消息发送了多遍（心跳时就会再发送），那么一条消息会导致n次上涨
        //同意了日志，就更新对应的m_matchIndex和m_nextIndex
        m_matchIndex[server] = std::max(m_matchIndex[server], args->prevlogindex() + args->entries_size());
        m_nextIndex[server] = m_matchIndex[server] + 1;
        int lastLogIndex = getLastLogIndex();

        myAssert(m_nextIndex[server] <= lastLogIndex + 1,
            format("error msg:rf.nextIndex[%d] > lastLogIndex+1, len(rf.logs) = %d   lastLogIndex{%d} = %d", server,
                m_logs.size(), server, lastLogIndex));
        if (*appendNums >= 1 + m_peers.size() / 2) {
            //可以commit了
            //两种方法保证幂等性，1.赋值为0 	2.上面≥改为==
            *appendNums = 0;

            // todo https://578223592-laughing-halibut-wxvpggvw69qh99q4.github.dev/ 不断遍历来统计rf.commitIndex
            //改了好久！！！！！
            // leader只有在当前term有日志提交的时候才更新commitIndex，因为raft无法保证之前term的Index是否提交
            //只有当前term有日志提交，之前term的log才可以被提交，只有这样才能保证“领导人完备性{当选领导人的节点拥有之前被提交的所有log，当然也可能有一些没有被提交的}”
            // rf.leaderUpdateCommitIndex()
            if (args->entries_size() > 0) {
                // DPrintf("args->entries(args->entries_size()-1).logterm(){%d}   m_currentTerm{%d}",
                //     args->entries(args->entries_size() - 1).logterm(), m_currentTerm);
                LOG << "args->entries(args->entries_size()-1).logterm(){"
                    << args->entries(args->entries_size() - 1).logterm() << "}   m_currentTerm{"
                    << m_currentTerm << "}";
            }
            if (args->entries_size() > 0 && args->entries(args->entries_size() - 1).logterm() == m_currentTerm) {
                // DPrintf(
                //     "---------------------------tmp------------------------- 当前term有log成功提交，更新leader的m_commitIndex "
                //     "from{%d} to{%d}",
                //     m_commitIndex, args->prevlogindex() + args->entries_size());
                LOG << "---------------------------tmp------------------------- 当前term有log成功提交，更新leader的m_commitIndex "
                    << "from{" << m_commitIndex << "} to{" << args->prevlogindex() + args->entries_size() << "}";

                m_commitIndex = std::max(m_commitIndex, args->prevlogindex() + args->entries_size());
            }
            myAssert(m_commitIndex <= lastLogIndex,
                format("[func-sendAppendEntries,rf{%d}] lastLogIndex:%d  rf.commitIndex:%d\n", m_me, lastLogIndex,
                    m_commitIndex));
            // fmt.Printf("[func-sendAppendEntries,rf{%v}] len(rf.logs):%v  rf.commitIndex:%v\n", rf.me, len(rf.logs),
            // rf.commitIndex)
        }
    }
    return ok;
}

void Raft::AppendEntries(google::protobuf::RpcController* controller,
    const ::raftRpcProctoc::AppendEntriesArgs* request,
    ::raftRpcProctoc::AppendEntriesReply* response, ::google::protobuf::Closure* done) {
    AppendEntries1(request, response);
    done->Run();
}

void Raft::InstallSnapshot(google::protobuf::RpcController* controller,
    const ::raftRpcProctoc::InstallSnapshotRequest* request,
    ::raftRpcProctoc::InstallSnapshotResponse* response, ::google::protobuf::Closure* done) {
    InstallSnapshot(request, response);

    done->Run();
}

void Raft::RequestVote(google::protobuf::RpcController* controller, const ::raftRpcProctoc::RequestVoteArgs* request,
    ::raftRpcProctoc::RequestVoteReply* response, ::google::protobuf::Closure* done) {
    RequestVote(request, response);
    done->Run();
}

void Raft::Start(Op command, int* newLogIndex, int* newLogTerm, bool* isLeader) {

    if (m_status != Leader) {
        // DPrintf("[func-Start-rf{%d}]  is not leader");
        LOG << "[func-Start-rf{" << m_me << "}]  is not leader";
        *newLogIndex = -1;
        *newLogTerm = -1;
        *isLeader = false;
        return;
    }

    std::lock_guard<std::mutex> lg1(m_mtx);

    raftRpcProctoc::LogEntry newLogEntry;
    newLogEntry.set_command(command.asString());
    newLogEntry.set_logterm(m_currentTerm);
    newLogEntry.set_logindex(getNewCommandIndex());
    m_logs.emplace_back(newLogEntry);

    int lastLogIndex = getLastLogIndex();


    persist();
    *newLogIndex = newLogEntry.logindex();
    *newLogTerm = newLogEntry.logterm();
    *isLeader = true;

    // 收到新日志后立即通知心跳线程
    {
        m_newLogAvailable = true; // 设置标志
        m_newLogCond.notify_one(); // 通知心跳线程
    }

}

// Make
// the service or tester wants to create a Raft server. the ports
// of all the Raft servers (including this one) are in peers[]. this
// server's port is peers[me]. all the servers' peers[] arrays
// have the same order. persister is a place for this server to
// save its persistent state, and also initially holds the most
// recent saved state, if any. applyCh is a channel on which the
// tester or service expects Raft to send ApplyMsg messages.
// Make() must return quickly, so it should start goroutines
// for any long-running work.
void Raft::init(std::vector<std::shared_ptr<RaftRpcUtil>> peers, int me, std::shared_ptr<Persister> persister,
    std::shared_ptr<LockQueue<ApplyMsg>> applyCh) {
    m_peers = peers; //与其他结点沟通的rpc类
    m_persister = persister; //持久化类
    m_me = me; //标记自己，毕竟不能给自己发送rpc吧
    m_threadPool = std::make_unique<ThreadPool>(peers.size()-1); // 初始化线程池，线程数可根据需求调整

    // Your initialization code here (2A, 2B, 2C).
    m_mtx.lock();

    // applier
    this->applyChan = applyCh; //与kv-server沟通
    //    rf.ApplyMsgQueue = make(chan ApplyMsg)
    m_currentTerm = 0; //初始化term为0
    m_status = Follower; //初始化身份为follower
    m_commitIndex = 0;
    m_lastApplied = 0;
    m_logs.clear();
    for (int i = 0; i < m_peers.size(); i++) {
        m_matchIndex.push_back(0);
        m_nextIndex.push_back(0);
    }
    m_votedFor = -1; //当前term没有给其他人投过票就用-1表示

    m_newLogCond;
    m_newLogAvailable = false; //标志位，是否有新的日志


    m_lastSnapshotIncludeIndex = 0;
    m_lastSnapshotIncludeTerm = 0;
    m_lastResetElectionTime = now();
    m_lastResetHearBeatTime = now();

    // initialize from state persisted before a crash

    readPersist(m_persister->ReadRaftState());
    if (m_lastSnapshotIncludeIndex > 0) {
        m_lastApplied = m_lastSnapshotIncludeIndex;
        // rf.commitIndex = rf.lastSnapshotIncludeIndex   todo ：崩溃恢复为何不能读取commitIndex
    }

    // DPrintf("[Init&ReInit] Sever %d, term %d, lastSnapshotIncludeIndex {%d} , lastSnapshotIncludeTerm {%d}", m_me,
    //     m_currentTerm, m_lastSnapshotIncludeIndex, m_lastSnapshotIncludeTerm);
    LOG << "[Init&ReInit] Sever {" << m_me << "}, term {" << m_currentTerm
        << "}, lastSnapshotIncludeIndex {" << m_lastSnapshotIncludeIndex
        << "} , lastSnapshotIncludeTerm {" << m_lastSnapshotIncludeTerm << "}";

    m_mtx.unlock();

    // m_ioManager = std::make_unique<monsoon::IOManager>(FIBER_THREAD_NUM, FIBER_USE_CALLER_THREAD);

    // start ticker fiber to start elections
    // 启动三个循环定时器
    // todo:原来是启动了三个线程，现在是直接使用了协程，三个函数中leaderHearBeatTicker
    // 、electionTimeOutTicker执行时间是恒定的，applierTicker时间受到数据库响应延迟和两次apply之间请求数量的影响，这个随着数据量增多可能不太合理，最好其还是启用一个线程。
    // m_ioManager->scheduler([this]() -> void { this->leaderHearBeatTicker(); });
    // m_ioManager->scheduler([this]() -> void { this->electionTimeOutTicker(); });
    // std::thread t3(&Raft::applierTicker, this);
    // t3.detach();

    std::thread t(&Raft::leaderHearBeatTicker, this);
    t.detach();

    std::thread t2(&Raft::electionTimeOutTicker, this);
    t2.detach();

    std::thread t3(&Raft::applierTicker, this);
    t3.detach();
}

std::string Raft::persistData() {
    BoostPersistRaftNode boostPersistRaftNode;
    boostPersistRaftNode.m_currentTerm = m_currentTerm;
    boostPersistRaftNode.m_votedFor = m_votedFor;
    boostPersistRaftNode.m_lastSnapshotIncludeIndex = m_lastSnapshotIncludeIndex;
    boostPersistRaftNode.m_lastSnapshotIncludeTerm = m_lastSnapshotIncludeTerm;
    for (auto& item : m_logs) {
        boostPersistRaftNode.m_logs.push_back(item.SerializeAsString());
    }

    std::stringstream ss;
    boost::archive::text_oarchive oa(ss);
    oa << boostPersistRaftNode;
    return ss.str();
}

void Raft::readPersist(std::string data) {
    if (data.empty()) {
        return;
    }
    std::stringstream iss(data);
    boost::archive::text_iarchive ia(iss);
    // read class state from archive
    BoostPersistRaftNode boostPersistRaftNode;
    ia >> boostPersistRaftNode;

    m_currentTerm = boostPersistRaftNode.m_currentTerm;
    m_votedFor = boostPersistRaftNode.m_votedFor;
    m_lastSnapshotIncludeIndex = boostPersistRaftNode.m_lastSnapshotIncludeIndex;
    m_lastSnapshotIncludeTerm = boostPersistRaftNode.m_lastSnapshotIncludeTerm;
    m_logs.clear();
    for (auto& item : boostPersistRaftNode.m_logs) {
        raftRpcProctoc::LogEntry logEntry;
        logEntry.ParseFromString(item);
        m_logs.emplace_back(logEntry);
    }
}

void Raft::Snapshot(int index, std::string snapshot) {
    std::lock_guard<std::mutex> lg(m_mtx);

    if (m_lastSnapshotIncludeIndex >= index || index > m_commitIndex) {
        LOG << "[func-Snapshot-rf{" << m_me << "}] rejects replacing log with snapshotIndex {" << index
            << "} as current snapshotIndex {" << m_lastSnapshotIncludeIndex << "} is larger or smaller ";
        return;
    }
    auto lastLogIndex = getLastLogIndex();  //为了检查snapshot前后日志是否一样，防止多截取或者少截取日志

    //制造完此快照后剩余的所有日志
    int newLastSnapshotIncludeIndex = index;
    int newLastSnapshotIncludeTerm = m_logs[getSlicesIndexFromLogIndex(index)].logterm();
    std::vector<raftRpcProctoc::LogEntry> trunckedLogs;
    // todo :这种写法有点笨，待改进，而且有内存泄漏的风险
    for (int i = index + 1; i <= getLastLogIndex(); i++) {
        //注意有=，因为要拿到最后一个日志
        trunckedLogs.push_back(m_logs[getSlicesIndexFromLogIndex(i)]);
    }
    m_lastSnapshotIncludeIndex = newLastSnapshotIncludeIndex;
    m_lastSnapshotIncludeTerm = newLastSnapshotIncludeTerm;
    m_logs = trunckedLogs;
    m_commitIndex = std::max(m_commitIndex, index);
    m_lastApplied = std::max(m_lastApplied, index);

    // rf.lastApplied = index //lastApplied 和 commit应不应该改变呢？？？ 为什么  不应该改变吧
    m_persister->Save(persistData(), snapshot);

    // DPrintf("[SnapShot]Server %d snapshot snapshot index {%d}, term {%d}, loglen {%d}", m_me, index,
    //     m_lastSnapshotIncludeTerm, m_logs.size());
    LOG << "[SnapShot]Server {" << m_me << "} snapshot snapshot index {" << index
        << "}, term {" << m_lastSnapshotIncludeTerm << "}, loglen {" << m_logs.size() << "}";
    myAssert(m_logs.size() + m_lastSnapshotIncludeIndex == lastLogIndex,
        format("len(rf.logs){%d} + rf.lastSnapshotIncludeIndex{%d} != lastLogjInde{%d}", m_logs.size(),
            m_lastSnapshotIncludeIndex, lastLogIndex));
}